package com.jthinking.service.kafka.delay.queue.storage;

import com.jthinking.service.kafka.delay.queue.config.StorageProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.*;

/**
 * 一个Block块为指定分钟内的所有过期值
 */
@Service
public class StorageService {

    private static final Logger LOGGER = LoggerFactory.getLogger(StorageService.class);

    private final Map<String, FileTemp> temp = new HashMap<>();

    @Resource
    private StorageProperties storageProperties;


    public void add(Long second, byte[] value) {
        Calendar instance = Calendar.getInstance();
        instance.setTimeInMillis(System.currentTimeMillis() + second * 1000);

        int year = instance.get(Calendar.YEAR);
        String month = String.format("%02d", instance.get(Calendar.MONTH) + 1);
        String day = String.format("%02d", instance.get(Calendar.DAY_OF_MONTH));
        String hour = String.format("%02d", instance.get(Calendar.HOUR_OF_DAY));
        String minute = String.format("%02d", instance.get(Calendar.MINUTE));
        String minuteStr = year + month + day + hour + minute;
        int secondIndex = instance.get(Calendar.SECOND);

        add(minuteStr, secondIndex, value);
    }

    /**
     *
     * @param minuteStr 202201061102
     * @param secondIndex 0-59
     * @param value 内容
     */
    public synchronized void add(String minuteStr, int secondIndex, byte[] value) {

        String filepath = storageProperties.getDataPath() + "/" + minuteStr.substring(0, 8);

        File path = new File(filepath);
        if (!path.exists()) {
            path.mkdirs();
        }
        String filename = filepath + "/" + minuteStr;
        File file = new File(filename);

        ByteBuffer byteBuffer = ByteBuffer.allocate(4 + 4 + value.length)
                .putInt(secondIndex)
                .putInt(value.length)
                .put(value);
        byteBuffer.flip();

        FileTemp fileTemp = temp.get(minuteStr);
        if (fileTemp == null) {
            try {
                RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw");
                FileChannel fc = randomAccessFile.getChannel();
                fc.write(byteBuffer, fc.size());
                temp.put(minuteStr, new FileTemp(randomAccessFile, fc));
            } catch (IOException e) {
                LOGGER.error("", e);
            }
        } else {
            try {
                FileChannel fc = fileTemp.getFileChannel();
                fc.write(byteBuffer, fc.size());
            } catch (IOException e) {
                LOGGER.error("", e);
            }
        }
    }

    /**
     *
     * @param minuteStr 202201061102
     * @param secondIndex 0-59
     */
    public List<byte[]> get(String minuteStr, int secondIndex) {
        List<byte[]> list = new ArrayList<>();
        FileTemp fileTemp = temp.get(minuteStr);
        if (fileTemp != null) {
            try {
                FileChannel fc = fileTemp.getFileChannel();
                ByteBuffer allocate = ByteBuffer.allocate((int) fc.size());
                fc.read(allocate, 0);
                allocate.flip();
                while (allocate.hasRemaining()) {
                    int second = allocate.getInt();
                    int len = allocate.getInt();
                    if (second == secondIndex) {
                        byte[] key = new byte[len];
                        allocate.get(key);
                        list.add(key);
                    } else {
                        int newPosition = allocate.position() + len;
                        allocate.position(Math.min(newPosition, allocate.limit()));
                    }
                }
                allocate.clear();
            } catch (IOException e) {
                LOGGER.error("", e);
            }
        }
        return list;
    }

    private void monitor() {
        Thread thread = new Thread(() -> {
            while (true) {

                for (Map.Entry<String, FileTemp> entry : temp.entrySet()) {
                    try {
                        String filename = entry.getKey();

                        // TODO 判断名称是否在范围之外，aaa

                        FileTemp fileTemp = entry.getValue();
                        fileTemp.close();
                        LOGGER.info("close file: {}", filename);
                    } catch (IOException e) {
                        LOGGER.error("", e);
                    }
                }

                try {
                    Thread.sleep(1000);
                } catch (Exception e) {
                    LOGGER.error("", e);
                    break;
                }
            }
        });
        thread.setDaemon(true);
        thread.setName("file temp clear");
        thread.start();
    }


    public static class FileTemp implements Closeable {
        private RandomAccessFile randomAccessFile;
        private FileChannel fileChannel;

        public FileTemp(RandomAccessFile randomAccessFile, FileChannel fileChannel) {
            this.randomAccessFile = randomAccessFile;
            this.fileChannel = fileChannel;
        }

        public RandomAccessFile getRandomAccessFile() {
            return randomAccessFile;
        }

        public void setRandomAccessFile(RandomAccessFile randomAccessFile) {
            this.randomAccessFile = randomAccessFile;
        }

        public FileChannel getFileChannel() {
            return fileChannel;
        }

        public void setFileChannel(FileChannel fileChannel) {
            this.fileChannel = fileChannel;
        }

        @Override
        public void close() throws IOException {
            if (fileChannel != null) {
                fileChannel.close();
            }
            if (randomAccessFile != null) {
                randomAccessFile.close();
            }
        }


    }





}
